package cluster.back

import akka.actor.{Actor, ActorSystem, Props, RootActorPath}
import akka.cluster.ClusterEvent.{CurrentClusterState, MemberUp}
import akka.cluster.{Cluster, Member, MemberStatus}
import cluster.{BackendRegistration, TransformationJob, TransformationResult}
import com.typesafe.config.ConfigFactory

/**
  * Created by shouf on 2016/5/19.
  */
class TransformationBackend extends Actor {
  val cluster = Cluster(context.system)

  // subscribe to cluster changes, MemberUp
  // re-subscribe when restart
  override def preStart(): Unit = cluster.subscribe(self, classOf[MemberUp])//cluster.subscribe订阅事件  注册此种actor能被加入到集群中
  override def postStop(): Unit = cluster.unsubscribe(self)

  def receive = {
    case TransformationJob(text) => sender() ! TransformationResult(text.toUpperCase)//模拟后台数据处理， 并再次发送到前台
    case state: CurrentClusterState =>
      state.members.filter(_.status == MemberStatus.Up) foreach register
    case MemberUp(m) => register(m)
  }

  def register(member: Member): Unit =
    if (member.hasRole("frontend")) {
      println(RootActorPath(member.address) / "user" / "frontend")
      context.actorSelection(RootActorPath(member.address) / "user" / "frontend") ! BackendRegistration
    }
}
object TransformationBackend extends App {
  // Override the configuration of the port when specified as program argument
  val port = if (args.isEmpty) "0" else args(0)
  val config = ConfigFactory.parseString(s"akka.remote.netty.tcp.port=$port").
    withFallback(ConfigFactory.parseString("akka.cluster.roles = [backend]")).
    withFallback(ConfigFactory.load().getConfig("cluster"))

  val system = ActorSystem("clusterSys", config)
  system.actorOf(Props[TransformationBackend], name = "backend")
}